package sparkSQL

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode, SparkSession}

import java.time.LocalDateTime
import java.time.format.{DateTimeFormatter, DateTimeParseException}
import java.util.Properties
import scala.collection.immutable.HashMap

case class Student(name:String,age:Int)

object Demo01_sqlGrammar {

    val props = new Properties()
    props.put("user","root")
    props.put("password","root")

  /**
   * 初始化SparkSession
   * @return
   */
  def getSparkSession(
     appName:String,
     master:String="local[*]",
     isHive:Boolean=false,
     other:HashMap[String,Any]=null
   ):SparkSession = {
    val spark: SparkSession.Builder = SparkSession.builder()
    spark.master(master).appName(appName)

    if(other!=null){
      other.foreach(prop=>{
        spark.config(prop._1,prop._2.toString)
      })
    }

    if (isHive) {
      spark.enableHiveSupport()
    }

    spark.getOrCreate()
  }



  /**
   * 读取原始数据
   * 1.如果读取的数据源没有任何格式上的指定，默认使用的是parquet列式存储文件
   *   如需读取指定格式的数据源，需要在read的后面添加format方法指定文件类型
   *   也可以把 format 和 load 都去掉，简化成以文件格式作为的方法名
   * 2.csv文件默认把所有行都当成数据处理，可以设置 option("header","true")
   *   把csv文件中的第一行当成表头
   *
   */
  def readData(spark:SparkSession) = {
//    val df:DataFrame = spark.read.format("json").load("D:/data/data_spark/people.json")
//    val df:DataFrame = spark.read.json("D:/data/data_spark/people.json")
//    val df:DataFrame = spark.read.option("header","true").csv("D:/data/data_spark/net/netClean_test.csv")

    //以普通方式读取jdbc的原始数据
//    val df:DataFrame = spark.read.format("jdbc")
//      .option("url","jdbc:mysql://localhost:3306/bookshop")
//      .option("dbtable","tb_book")
//      .option("user","root")
//      .option("password","root")
//      .load()

    //以简写的形式读取jdbc的数据

    val df:DataFrame = spark.read.jdbc(
      "jdbc:mysql://localhost:3306/bookshop",
      "tb_book",
      props
    )

//    df.show()
//    df.printSchema()

    df

  }

  /**
   * 写出数据
   * 1.如果写出的时候没有特殊指定，以parquet格式写出
   *   如果有特殊需求，需要通过format指定
   *   简写形式，直接调用文件类型的方法名
   * 2.保存模式
   *   1.ErrorIfExists  如果存在就报错
   *   2.Append         追加
   *   3.Overwrite      覆盖
   *   4.Ignore         忽略
   * @param df
   */
  def writeData(df:DataFrame) = {
//    df.write.format("csv").save("E://result")
    df.write.mode(SaveMode.Append).option("header","true").csv("E://result")

//    df.write.mode(SaveMode.ErrorIfExists).jdbc(
//      "jdbc:mysql://localhost:3306/taxi",
//      "result",
//      props
//    )

  }

  /**
   * 以DataSetAPI的形式处理数据
   * 所有的sql关键词都有相应的API对应
   * 使用起来比较简单，但是API需要熟悉
   * @param df
   */
  def processDataByAPI(spark:SparkSession,df:DataFrame) = {

    import spark.implicits._

    val ds:Dataset[Row] = df.select("book_name", "author", "publisher", "price")
//      .where("price<100")
      .where($"price"<100)
      .orderBy("price")

//    val ds:Dataset[Row] = df.select("publisher")
//      .groupBy("publisher")
//      .count()


    ds.show()
    ds.printSchema()
  }

  /**
   * 以SQL的形式处理数据
   * sql语句形式比较麻烦，但是是我们熟悉的东西
   * @param df
   */
  def processDataBySQL(spark:SparkSession,df:DataFrame) = {
    //1.普通临时视图，如果创建重名表会报错
//    df.createTempView("book")
    //2.可以做替换的临时视图，如果创建重名表不会报错
//    df.createOrReplaceTempView("book")
    //3.创建全局的临时视图,但是使用表的时候需要指定 global_temp.表
//    df.createGlobalTempView("book")
    //4.创建可以替换的临时视图，在全局的基础上如果创建重名表不会报错
//    df.createOrReplaceGlobalTempView("book")


    //创建持久化表：相当于是数据写入文件和创建表的整合
    df.write.mode(SaveMode.Overwrite).option("path","E://book").saveAsTable("book")


    //创建一个新的Session
    val sparkNew = spark.newSession()

    val sql =
      """
        |select book_name,author,publisher,price
        |from book
        |where price<100
        |order by price
        |""".stripMargin

    val ds: DataFrame = spark.sql(sql)

//    ds.show()


    //如果此文件只在当前sql语句中使用，可以在sql中直接读取文件中的数据
    val ds1 = spark.sql("select * from json.`D:/data/data_spark/people.json` where age<20")

    ds1.show()

  }


  /**
   * 普通集合转成DF/DS
   */
  def collection2DF(spark:SparkSession) = {

    import spark.implicits._

    val list_num = List(1,2,3,4,5,6,7,8,9,10)
    //把普通集合转成了DataSet,DataSet本身没有表头，会使用默认的表头
    val num_ds:Dataset[Int] = list_num.toDS()
    //把普通集合转成了DataFrame,如果数据中有表头就直接使用，如果没有使用默认的表头
    val num_df:DataFrame = list_num.toDF()
    val num_df_schema:DataFrame = list_num.toDF("number")

    num_df_schema.show()

    val list_student = List(("tom",10),("jerry",20),("jack",30))
    val student_df:DataFrame = list_student.toDF()
    val student_df_schema:DataFrame = list_student.toDF("name","age")
    student_df.show()
    student_df_schema.show()

    //定义一个字符串类型的集合，装成某个类的实例化对象的集合，数据就可以与类的形参自动匹配，形参就是表头
    val list_student_str = List("tom,10","jerry,20","jack,30")
    val objectDF = list_student_str.map(stuStr =>{
      val fields = stuStr.split(",")
      Student(fields(0),fields(1).toInt)
    })

    val object_ds:Dataset[Student] = objectDF.toDS()
    val object_df:DataFrame = objectDF.toDF()
    val object_df_schema:DataFrame = objectDF.toDF("n","a")

    object_ds.show()
    object_df.show()
    object_df_schema.show()

  }

  /**
   * 把RDD转成DataSet/DataFrame
   * 可以进行类型的自动推断
   * @param spark
   */
  def rdd2DF(spark:SparkSession) = {

    import spark.implicits._

    val sc:SparkContext = spark.sparkContext
    val list_student_str = List("tom,10","jerry,20","jack,30")
    val student_rdd:RDD[String] = sc.parallelize(list_student_str)

    val objectRDD:RDD[Student] = student_rdd.map(stuStr =>{
      val fields = stuStr.split(",")
      Student(fields(0),fields(1).toInt)
    })

    val object_ds:Dataset[Student] = objectRDD.toDS()
    val object_df:DataFrame = objectRDD.toDF()
    val object_df_schema:DataFrame = objectRDD.toDF("n","a")

    object_ds.show()
    object_df.show()
    object_df_schema.show()


  }

  /**
   * 把RDD转成DataSet/DataFrame 可以自定义表头
   */
  def rdd2DF_schema(spark:SparkSession) = {

    val sc:SparkContext = spark.sparkContext
    val list_student_str = List("tom,10","jerry,20","jack,30")
    val student_rdd:RDD[String] = sc.parallelize(list_student_str)

    //把RDD转成一行要表示的所有列的Row类型
    val student_rdd_row:RDD[Row] = student_rdd.map(student=>{
      val fields = student.split(",")
      Row(fields(0),fields(1).toInt)
    })

    //定义出所有需要的表头，以字符串形式表示
    //根据实际情况对每个字段进行封装，形式任意，只要最终的结果一致就可以
    val title = "name,age"
    val fields:Array[StructField] = title.split(",").map(fieldName=>{
      if(fieldName.equals("age")){
        StructField(fieldName,IntegerType,false)
      }else{
        StructField(fieldName,StringType,true)
      }
    })

    val schema:StructType = StructType(fields)

    val stuDF:DataFrame = spark.createDataFrame(student_rdd_row,schema)

    stuDF.show()
    stuDF.printSchema()
  }

  /**
   * 把df转成rdd
   * @param spark
   */
  def df2RDD(spark:SparkSession) = {

    val netDF: DataFrame = spark.read.option("header","true").csv("D:/data/data_spark/net/netClean_test.csv")

    val netRDD:RDD[Row] = netDF.rdd

    netRDD.map(row=>{
      val userId = row(0)
      val startTime = row(1)
      val endTime = row(2)

      userId+"#"+startTime+"#"+endTime

    }).foreach(println)


  }

  /**
   * 使用 sparkCore 和 sparkSQL 整合处理清洗上网数据案例
   * @param spark
   */
  def cleanNetData(spark:SparkSession) = {

    import spark.implicits._

    val netDF: DataFrame = spark.read.option("header","true").csv("D:/data/data_spark/net/netClean_test.csv")
    val sexDF: DataFrame = spark.read.option("header","true").csv("D:/data/data_spark/net/sexDictFile.csv")

    netDF.createOrReplaceTempView("net")
    sexDF.createOrReplaceTempView("sex")

    val sql =
      """
        |select n.userId,gender,startTime,endTime
        |from net as n
        |left join sex as s
        |on n.userId=s.userId
        |where n.userId is not null and startTime is not null and endTime is not null
        |and unix_timestamp(startTime,"yyyy/M/d H:m") < unix_timestamp(endTime,"yyyy/M/d H:m")
        |""".stripMargin

    val df = spark.sql(sql)

//    val netCleanedRDD:RDD[Row] = df.rdd
//
//    val resultRDD:RDD[(String,String,String,String)] = netCleanedRDD.filter(row=>{
//      var flag = true
//      val startTime:String = row(2).toString
//      val endTime:String = row(3).toString
//      var startTime_date:LocalDateTime = null
//      var endTime_date:LocalDateTime = null
//      try{
//        startTime_date = LocalDateTime.parse(startTime, DateTimeFormatter.ofPattern("yyyy/M/d H:m"))
//        endTime_date = LocalDateTime.parse(endTime, DateTimeFormatter.ofPattern("yyyy/M/d H:m"))
//      }catch {
//        case e:DateTimeParseException => flag = false
//      }
//      flag && startTime_date.isBefore(endTime_date)
//    }).map(row=>{
//      (row(0).toString,row(1).toString,row(2).toString,row(3).toString)
//    })
//
//    val resultDF:DataFrame = resultRDD.toDF("userId","gender","startTime","endTime")
//
//    resultDF.show(40)

    df.show()

  }


  /**
   * sparkSQL 对于 Hive 的支持
   * 如果没有任何的特殊指定的情况下，Spark会在当前工程目录下创建元数据库(derby)和数据仓库
   * @param spark
   */
  def hiveSupport() = {

//    val propsMap = HashMap(("spark.sql.warehouse.dir","E://warehouse"))

    val spark: SparkSession = getSparkSession(appName = "hive", isHive = true)

    val df = spark.sql("show tables")

    df.show()
  }

  /**
   * 分区分桶排序
   * @param spark
   */
  def bucketPartitionSort(spark:SparkSession) = {

    val df = spark.read.option("header","true").csv("D:/data/data_spark/net/netClean.csv")

    df.write
      .mode(SaveMode.Append)
      .format("csv")
      .option("header","true")
      .partitionBy("userId")
      .bucketBy(3,"startTime")
      .sortBy("endTime")
      .option("path","file:///E://net")
      .saveAsTable("net")

    val resultDF = spark.sql("select * from net")

    resultDF.show(1000)

  }


  def main(args: Array[String]): Unit = {
    val spark: SparkSession = getSparkSession("sql")

//    val df:DataFrame = readData(spark)
//    processDataByAPI(spark,df)
//    processDataBySQL(spark,df)

//    collection2DF(spark)
//    rdd2DF(spark)
//    rdd2DF_schema(spark)
//    df2RDD(spark)
//    cleanNetData(spark)
//    hiveSupport()
    bucketPartitionSort(spark)

//    writeData(df)

    spark.close()
  }

}
